package com.ruyuan.rapid.core.netty.processor;

import com.lmax.disruptor.dsl.ProducerType;
import com.ruyuan.rapid.common.concurrent.queue.flusher.ParallelFlusher;
import com.ruyuan.rapid.common.enums.ResponseCode;
import com.ruyuan.rapid.core.RapidConfig;
import com.ruyuan.rapid.core.context.HttpRequestWrapper;
import com.ruyuan.rapid.core.helper.ResponseHelper;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.http.*;
import lombok.extern.slf4j.Slf4j;

/**
 * 缓冲队列的核心实现, 最终调用的方法还是通过NettyCoreProcessor
 * @author mac
 */
@Slf4j
public class NettyBatchEventProcessor implements NettyProcessor {

    private static final String THREAD_NAME_PREFIX = "rapid-flusher-";

    private RapidConfig rapidConfig;

    private NettyCoreProcessor nettyCoreProcessor;

    private ParallelFlusher<HttpRequestWrapper> parallelFlusher;

    public NettyBatchEventProcessor(RapidConfig rapidConfig, NettyCoreProcessor nettyCoreProcessor) {
        this.rapidConfig = rapidConfig;
        this.nettyCoreProcessor = nettyCoreProcessor;
        ParallelFlusher.Builder<HttpRequestWrapper> builder = new ParallelFlusher.Builder<>();
        builder.setBufferSize(rapidConfig.getBufferSize())
                .setThreads(rapidConfig.getProcessThread())
                .setProducerType(ProducerType.MULTI)
                .setNamePrefix(THREAD_NAME_PREFIX)
                .setWaitStrategy(rapidConfig.getActualWaitStrategy());
        BatchEventProcessorListener batchEventProcessorListener = new BatchEventProcessorListener();
        builder.setEventListener(batchEventProcessorListener);
        parallelFlusher = builder.build();
    }

    /**
     * 核心执行方法
     * @param httpRequestWrapper
     */
    @Override
    public void process(HttpRequestWrapper httpRequestWrapper) {
        System.out.println("NettyBatchEventProcessor add ok......");
        this.parallelFlusher.add(httpRequestWrapper);
    }

    /**
     * 执行器启动方法
     */
    @Override
    public void start() {
        this.nettyCoreProcessor.start();
        this.parallelFlusher.start();
    }

    /**
     * 执行器关闭方法,资源释放
     */
    @Override
    public void shutdown() {
        this.nettyCoreProcessor.shutdown();
        this.parallelFlusher.shutdown();
    }

    /**
     * 监听事件的核心处理逻辑
     */
    public class BatchEventProcessorListener implements ParallelFlusher.EventListener<HttpRequestWrapper> {
        /**
         * 成功时调用
         * @param event
         * @throws Exception
         */
        @Override
        public void onEvent(HttpRequestWrapper event) throws Exception {
            nettyCoreProcessor.process(event);
        }

        /**
         * 失败时处理异常
         * @param t
         * @param sequence
         * @param event
         */
        @Override
        public void onException(Throwable t, long sequence, HttpRequestWrapper event) {
            HttpRequest request = event.getFullHttpRequest();
            ChannelHandlerContext ctx = event.getCtx();
            try {
                log.error("BatchEventProcessorListener onException 请求处理失败, request: {}, errorMessage: {}", request, t.getMessage(), t);
                FullHttpResponse fullHttpResponse = ResponseHelper.getHttpResponse(ResponseCode.INTERNAL_ERROR);

                if (!HttpUtil.isKeepAlive(request)) {
                    ctx.writeAndFlush(fullHttpResponse).addListener(ChannelFutureListener.CLOSE);
                } else {
                    fullHttpResponse.headers().set(HttpHeaderNames.CONNECTION, HttpHeaderValues.KEEP_ALIVE);
                    ctx.writeAndFlush(fullHttpResponse);
                }
            } catch (Exception e) {
                // ignore
                log.error("BatchEventProcessorListener onException 请求回写失败, request: {}, errorMessage: {}", request, e.getMessage(), e);
            }
        }
    }
}
